package com.atguigu.gmall.realtime.common.base;

import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.FlinkSqlUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION;

public abstract class BaseSqlApp {
    public void start(int port, int parallelism, String ckPath){
        // 准备执行环境
        // TableEnvironment
        // StreamTableEnvironment
        Configuration conf = new Configuration();
        conf.setInteger(RestOptions.PORT, port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(parallelism);


        // 检查点配置
        // 2.状态后端&检查点相关设置
        // 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        // 开启 checkpoint
        env.enableCheckpointing(5000);
        // 设置 checkpoint 模式: 精准一次
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // checkpoint 存储
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall2023/stream/" + ckPath);
        // checkpoint 并发数
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // checkpoint 之间的最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
        // checkpoint  的超时时间
        env.getCheckpointConfig().setCheckpointTimeout(10000);
        // job 取消时 checkpoint 保留策略
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(RETAIN_ON_CANCELLATION);


        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
        // 操作hdfs的用户
        System.setProperty("HADOOP_USER_NAME", Constant.HDFS_USER_NAME);

        // 流表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 读取数据 + 处理数据 + 写出数据
        handle(tableEnv, env);

    }


    public abstract  void handle(StreamTableEnvironment tableEnv, StreamExecutionEnvironment env);

    public void readOdsTopicDb(StreamTableEnvironment tableEnv, String groupId){
        String sourceTable = "CREATE TABLE topic_db (" +
                "  `database` STRING," +
                "  `table` STRING," +
                "  `type` STRING," +
                "  `ts` BIGINT, " +
                "  `data` MAP<STRING,STRING>, " +
                " `old` MAP<STRING,STRING>," +
                " `pt` as PROCTIME(), " +
                " `et` as to_timestamp_ltz(ts,0), " +
                " watermark for et as et - interval '3' second " +
                ")" + FlinkSqlUtil.getKafkaSourceDDL(Constant.TOPIC_DB, groupId);
        tableEnv.executeSql(sourceTable);
    }

    public  static void readDimBaseDic(StreamTableEnvironment tableEnv) {
        String createBaseDicSql =
                " create table dim_base_dic (" +
                        " dic_code STRING , " +
                        " info ROW<dic_name STRING> , " +
                        " PRIMARY KEY (dic_code) NOT ENFORCED " +
                        ") " + FlinkSqlUtil.getHbaseDDL(Constant.HBASE_NAMESPACE , "dim_base_dic");

        tableEnv.executeSql(createBaseDicSql);
    }



}
